package com.zy.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.bean.BaseDbTableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

public class MyBaseAppConnect extends BroadcastProcessFunction<JSONObject, String, JSONObject>{
    MapStateDescriptor<String,BaseDbTableProcess> mapStateDescriptor;//广播流状态

    Map<String,BaseDbTableProcess> mapConfig = new HashMap<>();

    public MyBaseAppConnect(MapStateDescriptor<String,BaseDbTableProcess> mapStateDescriptor){
        this.mapStateDescriptor = mapStateDescriptor;//传入状态
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //预加载
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //获取数据库对象
        String sql = "select * from gmall_config.table_process_dwd ";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        //处理
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()){
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i < metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object object = rs.getObject(i);
                jsonObj.put(columnName,object);
            }

            BaseDbTableProcess baseDbTableProcess = jsonObj.toJavaObject(BaseDbTableProcess.class);
            mapConfig.put(baseDbTableProcess.getSinkTable() + ":" + baseDbTableProcess.getSourceType(), baseDbTableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        // {"database":"gmall","xid":41793,"data":{"create_time":"2022-06-08 10:40:45","user_id":752,"sku_id":31,"id":1608723429735805940,"spu_id":10,"is_cancel":"1"},"old":{"is_cancel":"0"},"commit":true,"type":"update","table":"favor_info","ts":1693020818}
        //获取key = 表明 ： 类型名
        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");
        String key = table + ":" + type;
        //判断是否是状态中表
            //获取状态
        ReadOnlyBroadcastState<String, BaseDbTableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);//获取状态
        BaseDbTableProcess baseDbTableProcess = null;
        baseDbTableProcess = broadcastState.get(key);

        //如果是进行处理
        if (baseDbTableProcess != null || (baseDbTableProcess = mapConfig.get(table + ":" + type)) != null){
            //不为空的话，说明在状态中获取到了数据,对该json数据进行处理
            //获取到data JSON
            JSONObject data = jsonObject.getJSONObject("data");
            //去掉多余列
            String sinkColumns = baseDbTableProcess.getSinkColumns();
                //过滤
            filterColumn(data,sinkColumns);
            //增加需要的列 ts 输出的sink
            String sinkTable = baseDbTableProcess.getSinkTable();
            data.put("sinkTable",sinkTable);
            Long ts = jsonObject.getLong("ts");//作为事件时间
            data.put("ts",ts);
            collector.collect(data);
        }
    }

    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] split = sinkColumns.split(",");
        //date中有多余的数据， sink中是需要的数据  判断是否在sink中 不在就删除
        List<String> list = Arrays.asList(split);
        Set<Map.Entry<String, Object>> entrySet = data.entrySet();
        entrySet.removeIf(new Predicate<Map.Entry<String, Object>>() {
            @Override
            public boolean test(Map.Entry<String, Object> stringObjectEntry) {
                return !list.contains(stringObjectEntry.getKey());
            }
        });

    }

    @Override
    public void processBroadcastElement(String s, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //"op":"r":  {"before":null,"after":{"source_table":"user_info","sink_table":"dim_user_info","sink_family":"info","sink_columns":"id,login_name,name,user_level,birthday,gender,create_time,operate_time","sink_row_key":"id"},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1692426976215,"transaction":null}
        //"op":"d": {"before":{"source_table":"aa","sink_table":"dim_a","sink_family":"info","sink_columns":"id,name,age","sink_row_key":"id"},"after":null,"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1692427178000,"snapshot":"false","db":"gmall0315_config","sequence":null,"table":"table_process_dim","server_id":1,"gtid":null,"file":"mysql-bin.000004","pos":10908422,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1692427178485,"transaction":null}
        JSONObject jsonObj = JSONObject.parseObject(s);
        String op = jsonObj.getString("op");
        BroadcastState<String,BaseDbTableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);//获取状态
        //这个是为了时刻关注到 配置信息表中的数据变化
        if (op.equals("d")){
            //如果删除了配置表中的信息，就要相应的删除状态中的信息
            BaseDbTableProcess before = jsonObj.getObject("before", BaseDbTableProcess.class);
            String sourceTable = before.getSourceTable();
            String sourceType = before.getSourceType();
            String key = sourceTable + ":" + sourceType;
            broadcastState.remove(key);//按照key值删除状态
            mapConfig.remove(key);
        }else {
            //状态新增
            BaseDbTableProcess after = jsonObj.getObject("after", BaseDbTableProcess.class);
            String sourceTable = after.getSourceTable();
            String sourceType = after.getSourceType();
            String key = sourceTable + ":" + sourceType;
            broadcastState.put(key, after);
            mapConfig.put(key,after);
        }

    }
}
